package com.isyscore.os.metadata.kettle.step;

import com.isyscore.os.metadata.enums.KettleDataFlowNodeType;
import com.isyscore.os.metadata.kettle.base.FlowConfig;
import com.isyscore.os.metadata.kettle.base.Step;
import com.isyscore.os.metadata.kettle.vis.StreamLookupNode;
import com.isyscore.os.metadata.kettle.vis.VisNode;
import org.pentaho.di.core.database.DatabaseMeta;
import org.pentaho.di.core.plugins.PluginInterface;
import org.pentaho.di.core.plugins.PluginRegistry;
import org.pentaho.di.core.plugins.StepPluginType;
import org.pentaho.di.core.row.ValueMetaInterface;
import org.pentaho.di.trans.TransMeta;
import org.pentaho.di.trans.step.StepMeta;
import org.pentaho.di.trans.step.StepMetaInterface;
import org.pentaho.di.trans.step.errorhandling.StreamInterface;
import org.pentaho.di.trans.steps.streamlookup.StreamLookupMeta;

import java.util.List;

/**
 * @author dc
 * @Type StreamLookupStep.java
 * @Desc 流检测
 * @date 2022/10/20 15:57
 */
public class StreamLookupStep implements Step {
    @Override
    public StepMeta decode(VisNode step, List<DatabaseMeta> databases, TransMeta transMeta, FlowConfig transGraph) throws Exception {
        StreamLookupNode node = (StreamLookupNode) step;
        PluginRegistry registry = PluginRegistry.getInstance();
        PluginInterface sp = registry.findPluginWithId(StepPluginType.class, KettleDataFlowNodeType.StreamLookup.name());
        StepMetaInterface stepMetaInterface = (StepMetaInterface) registry.loadClass(sp);
        StreamLookupMeta streamLookupMeta = (StreamLookupMeta) stepMetaInterface;

        StreamInterface infoStream = streamLookupMeta.getStepIOMeta().getInfoStreams().get( 0 );
        StepMeta from = new StepMeta();
        from.setName(node.getLookupStepName());
        infoStream.setStepMeta(from);
        List<StreamLookupNode.JoinField> joinFields = node.getJoinFields();
        String[] keystream = new String[joinFields.size()];
        String[] keylookup = new String[joinFields.size()];
        for (int i = 0; i < joinFields.size(); i++) {
            StreamLookupNode.JoinField joinField = joinFields.get(i);
            keylookup[i]=joinField.getLookKey();
            keystream[i]=joinField.getKey();
        }
        streamLookupMeta.setKeylookup(keylookup);
        streamLookupMeta.setKeystream(keystream);
        List<StreamLookupNode.ReceiveField> receiveFields = node.getReceiveFields();
        String[] value = new String[receiveFields.size()];
        String[] rName = new String[receiveFields.size()];
        String[] valueDefault = new String[receiveFields.size()];
        int[] valueDefaultType = new int[receiveFields.size()];
        for (int i = 0; i < receiveFields.size(); i++) {
            StreamLookupNode.ReceiveField receiveField = receiveFields.get(i);
            valueDefaultType[i] = ValueMetaInterface.TYPE_NONE;
            valueDefault[i] = null;
            value[i] = receiveField.getValue();
            rName[i] = receiveField.getRenameValue();
        }
        streamLookupMeta.setValue(value);
        streamLookupMeta.setValueName(rName);

        streamLookupMeta.setValueDefault(valueDefault);
        streamLookupMeta.setValueDefaultType(valueDefaultType);
        StepMeta stepMeta = new StepMeta(KettleDataFlowNodeType.StreamLookup.name(), node.getNodeName(), stepMetaInterface);
        stepMeta.setDraw(true);
        return stepMeta;
    }

    @Override
    public StepMeta after(VisNode step, List<DatabaseMeta> databases, TransMeta transMeta, FlowConfig transGraph) throws Exception {
        return null;
    }
}
    
